//#include <iostream>
#include <errno.h>

#include "hbase_client.h"
#include "error_code.h"
#include "base.h"

using namespace std;

ColumnValue::ColumnValue()
{
	next = NULL;
}

ColumnValue::~ColumnValue()
{
}

RowResult::RowResult()
{
	index = NULL;
	cv = NULL;
	next = NULL;
}

RowResult::~RowResult()
{
}

void RowResult::free()
{
	ColumnValue *root, *n;
	root = cv;
	while (root) {
		n = root;
		root = root->next;
		delete (n);
	}
	cv = NULL;
	index = NULL;
	next = NULL;
	return;
}

int RowResult::insert(ColumnValue &_cv)
{
	ColumnValue *pcv = new ColumnValue;

	pcv->column = _cv.column;
	pcv->value = _cv.value;
	pcv->timestamp = _cv.timestamp;

	pcv->next = cv;
	cv = pcv;
	index = cv;

	return 0;
}

int RowResult::insert(ColumnValue *_cv)
{
	ColumnValue *pcv = new ColumnValue;

	pcv->column = _cv->column;
	pcv->value = _cv->value;
	pcv->timestamp = _cv->timestamp;

	pcv->next = cv;
	cv = pcv;
	index = cv;

	return 0;
}

int RowResult::insert(hbase::thrift::Text c, hbase::thrift::Text v, int64_t t)
{	
	ColumnValue *pcv = new ColumnValue;

	pcv->column = c;
	pcv->value = v;
	pcv->timestamp = t;

	pcv->next = cv;
	cv = pcv;
	index = cv;

	return 0;
}

ColumnValue *RowResult::nextcolumn()
{
	ColumnValue *pcv;

	if (index == NULL) {
		rotate();
		return NULL;
	}
	pcv = index;
	index = index->next;
	return pcv;
}

void RowResult::rotate()
{
	index = cv;
}

HbaseOp::HbaseOp(char *_sv_ip, unsigned short _sv_port)
{
	*sv_ip = '\0';
	strncpy(sv_ip, _sv_ip, INET_ADDRSTRLEN);
	sv_port = _sv_port;
	hclient = NULL;
}

HbaseOp::~HbaseOp()
{
	close();
}

int HbaseOp::delete_table(const char *tname)
{
    if (tname == NULL) {
        DERROR("hbase delete table argument illeagal\n");
        return -1;
    }
    
    hbase::thrift::Text table(tname);
    try {
        hclient->disableTable(tname); 
        hclient->deleteTable(table);
    } catch (hbase::thrift::IOError &e) {
        errnum = ERROR_CODE_IOERROR;
        DERROR("hbase delete table error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    } catch (TTransportException &e) {
		if (e.getType() == TTransportException::NOT_OPEN) {
			errnum = ERROR_CODE_SOCKET_CLOSED;
		} else {
			errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		}
        DERROR("hbase delete table error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase delete table error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    }
/*
    } catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase delete table error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
        return -1;
    }
*/
    return 0;
}

int HbaseOp::create(const char *tname, struct ColumnValue *cfamily, int compression)
{
	if (tname == NULL || cfamily == NULL) {
		DERROR("hbase create argument illeagal\n");
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
		return -1;
	}
	vector<hbase::thrift::ColumnDescriptor> col;
	hbase::thrift::ColumnDescriptor clmn;

	while (cfamily) {
		clmn.name = cfamily->column;
		if (compression)
			clmn.compression = "gz";
		col.push_back(clmn);
		cfamily = cfamily->next;
	}

	hbase::thrift::Text table(tname);
	try {
		hclient->createTable(table, col);
	} catch (hbase::thrift::AlreadyExists &e) {
		errnum = ERROR_CODE_ALREADY_EXISTS;
        DERROR("hbase create error: errrno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (hbase::thrift::IllegalArgument &e) {
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase create error: errrno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
	} catch (hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase create error: errrno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase create error: errrno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase create error: errrno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    }
/*
	} catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase create error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return -1;
	}
*/

	return 0;
}

RowResult *HbaseOp::get(const char *tname)
{
	return NULL;
}

RowResult *HbaseOp::get(const char *tname, const char *row)
{
	if (tname == NULL || row == NULL) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
		DERROR("hbase get argument illegal: %d\n", ERROR_CODE_ILLEGAL_ARGUMENT);
		return NULL;
	}
	vector<hbase::thrift::TRowResult> result;
	try {
		hclient->getRow(result, tname, row);
	} catch (hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;
	} catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return NULL;
    }
/*
	} catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		DERROR("hbase get error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return NULL;
	}
*/

	RowResult *rowret = new RowResult;
	rowret->table = tname;
	rowret->row = row;
	if (result.size() > 0) {
		std::map<hbase::thrift::Text, hbase::thrift::TCell>::iterator iter;

		for (size_t i = 0; i < result.size(); i++) {
			for (iter = result[i].columns.begin(); iter != result[i].columns.end(); iter++) {
				rowret->insert(iter->first, iter->second.value, iter->second.timestamp);
			}
		}
	} else {
		errnum = ERROR_CODE_EMPTY;
		delete(rowret);
		return NULL;
	}

	return rowret;
}

RowResult *HbaseOp::get(const char *tname, const char *row, const char *col)
{
	if (tname == NULL || row == NULL || col == NULL) {
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase get argument illegal: %d\n", ERROR_CODE_ILLEGAL_ARGUMENT);
		return NULL;
	}
	vector<hbase::thrift::TCell> result;
	try {
		hclient->get(result, tname, row, col);
	} catch (hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;
	} catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return NULL;
    }
/*
	} catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		DERROR("hbase get error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return NULL;
	}
*/

	RowResult *rowret = new RowResult;
	rowret->table = tname;
	rowret->row = row;
	if (result.size() > 0) {
		for (size_t i = 0; i < result.size(); i++)
			rowret->insert(col, result[i].value, result[i].timestamp);
	} else {
		errnum = ERROR_CODE_EMPTY;
		delete(rowret);
		return NULL;
	}

	return rowret;
}

RowResult *HbaseOp::get(const char *tname, const char *row, struct ColumnValue *columns)
{
	if (tname == NULL || row == NULL || columns == NULL) {
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase get argument illegal: %d\n", ERROR_CODE_ILLEGAL_ARGUMENT);
		return NULL;
	}
	vector<hbase::thrift::TRowResult> result;
	vector<hbase::thrift::Text> cols;
	while (columns) {
		cols.push_back(columns->column);
		columns = columns->next;
	}

	try {
		hclient->getRowWithColumns(result, tname, row, cols);
	} catch (hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;
	} catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase get error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return NULL;
    }
/*
	} catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		DERROR("hbase get error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return NULL;
	}
*/

	RowResult *rowret = new RowResult;
	rowret->table = tname;
	rowret->row = row;
	if (result.size() > 0) {
		std::map<hbase::thrift::Text, hbase::thrift::TCell>::iterator iter;
		for (size_t i = 0; i < result.size(); i++) {
			for (iter = result[i].columns.begin(); iter != result[i].columns.end(); iter++) {
				rowret->insert(iter->first, iter->second.value, iter->second.timestamp);
			}
		}
	} else {
		errnum = ERROR_CODE_EMPTY;
		delete(rowret);
		return NULL;
	}

	return rowret;
}

int HbaseOp::put(const char *tname, const char *row, struct RowResult *rowvalue)
{
	if (tname == NULL || row == NULL || rowvalue == NULL) {
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase put argument illegal: %d\n", ERROR_CODE_ILLEGAL_ARGUMENT);
		return -1;
	}
	vector<hbase::thrift::Mutation> c;
	//int i;
	ColumnValue *cv;

	while ((cv = rowvalue->nextcolumn()) != NULL) {
		hbase::thrift::Mutation m;
		m.column = cv->column;
		m.value = cv->value;
		c.push_back(m);
	}

	try {
		hclient->mutateRow(tname, row, c);
	} catch (hbase::thrift::IllegalArgument &e) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase put error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase put error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase put error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase put error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    }
/*
	} catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		DERROR("hbase put error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return -1;
	}
*/

	return 0;
}

int HbaseOp::put(const char *tname, struct RowResult *rows)
{
    if (tname == NULL || rows == NULL) {
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase put argument illegal: %d\n", ERROR_CODE_ILLEGAL_ARGUMENT);
        return -1;
    }
	vector<hbase::thrift::BatchMutation> v_rows_batch;
	struct RowResult *row = rows;
	while (row) {
		hbase::thrift::BatchMutation row_batch;
		row_batch.row = row->row;
		ColumnValue *cv;
		while ((cv = row->nextcolumn()) != NULL) {
			hbase::thrift::Mutation m;
			m.column = cv->column;
			m.value = cv->value;
			row_batch.mutations.push_back(m);
		}
		v_rows_batch.push_back(row_batch);
		row = row->next;
	}
    try {
        hclient->mutateRows(tname, v_rows_batch);
    } catch (hbase::thrift::IllegalArgument &e) {
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase put error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    } catch (hbase::thrift::IOError &e) {
        errnum = ERROR_CODE_IOERROR;
        DERROR("hbase put error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    } catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase put error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;

	} catch (TException &e) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase put error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	}
/*
    } catch (...) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase put error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
        return -1;
    }
*/

    return 0;
}

int HbaseOp::del(const char *tname, const char *row, struct ColumnValue *rowvalue)
{
	if (tname == NULL || row == NULL || rowvalue == NULL ||
		*tname == '\0' || *row == '\0'){
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase del argument illegal: %d\n", ERROR_CODE_ILLEGAL_ARGUMENT);
		return -1;
	}

	vector<hbase::thrift::Mutation> c;
	ColumnValue *cv = rowvalue;

	hbase::thrift::Mutation m;
	while (cv != NULL) {
		m.column = cv->column;
		//m.value = cv->value;
		m.isDelete = true;
		c.push_back(m);
		cv = cv->next;
	}

	try {
		hclient->mutateRow(tname, row, c);
	} catch (hbase::thrift::IllegalArgument &e) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase del error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase del error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase del error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase del error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    }
/*
	} catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		DERROR("hbase del error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return -1;
	}
*/

	return 0;
}
int HbaseOp::del(const char *tname, const char *row)
{
	if (tname == NULL || row == NULL ||
		*tname == '\0' || *row == '\0') {
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase del argument illegal: %d\n", ERROR_CODE_ILLEGAL_ARGUMENT);
		return -1;
	}
	try {
		hclient->deleteAllRow(tname, row);
	} catch (hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase del error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (hbase::thrift::IllegalArgument &e) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase del error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (TTransportException &e) {
		DERROR("del TTransportException: %d\n", e.getType());
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase del error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase del error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    }
/*
	} catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		DERROR("hbase del error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return -1;
	}
*/

	return 0;
}
RowResult *HbaseOp::scan(const int32_t nrows)
{
	RowResult *root = NULL;
	RowResult *tail = NULL;
	if (scanid < 0 || nrows <= 0) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
		DERROR("hbase scan error: must open first or argumen illegal\n");
		return NULL;
	}

	std::vector<hbase::thrift::TRowResult> rows;
	try {
		hclient->scannerGetList(rows, scanid, nrows);
	} catch(hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase scan error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;
	} catch (hbase::thrift::IllegalArgument &e) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase scan error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;
	} catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase scan error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return NULL;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase scan error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return NULL;
    }
/*
	} catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		DERROR("hbase scan error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return NULL;
	}
*/

	for (size_t i = 0; i < rows.size(); i++) {
		RowResult *row = new (struct RowResult);
		row->row = rows[i].row;

		std::map<hbase::thrift::Text, hbase::thrift::TCell>::iterator iter;
        for (iter = rows[i].columns.begin(); iter != rows[i].columns.end(); iter++) {
                row->insert(iter->first, iter->second.value, iter->second.timestamp);
        }

		if (root == NULL) {
			root = row;
		} else {
			tail->next = row;
		}
		tail = row;
	}
	if (root == NULL)
		errnum = ERROR_CODE_EMPTY;

	return root;
}

int HbaseOp::scanopen(const char *tname, const char *startrow, struct ColumnValue *colfamily)
{
	if (tname == NULL || startrow == NULL || colfamily == NULL) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
		DERROR("hbase scan open error: argumen illegal\n");
		return -1;
	}
	std::vector<std::string> columns;
	while (colfamily != NULL) {
		columns.push_back(colfamily->column);
		colfamily = colfamily->next;
	}

	try {
		scanid = hclient->scannerOpen(tname, startrow, columns);
	} catch (hbase::thrift::IOError &e) {
		scanid = -1;
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase scan open error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch(TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
		scanid = -1;
        DERROR("hbase scan open error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase scan open error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    }
/*
	} catch(...) {
		scanid = -1;
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
		DERROR("hbase scan open error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
		return -1;
	}
*/

	return 0;
}

int HbaseOp::scanOpenWithPrefix(const char *tname, const char *startrow, struct ColumnValue *colfamily)
{ 
    if (tname == NULL || startrow == NULL || colfamily == NULL) {
        errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase scan open error: argumen illegal\n");
        return -1;
    }
    std::vector<std::string> columns;
    while (colfamily != NULL) {
        columns.push_back(colfamily->column);
        colfamily = colfamily->next;
    }
  
    try {
        scanid = hclient->scannerOpenWithPrefix(tname, startrow, columns);
    } catch (hbase::thrift::IOError &e) {
        scanid = -1;
        errnum = ERROR_CODE_IOERROR; 
        DERROR("hbase scan open error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    } catch(TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        scanid = -1;
        DERROR("hbase scan open error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase scan open error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    }
/*
    } catch(...) {
        scanid = -1;
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase scan open error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
        return -1;
    }
*/

    return 0;
} 

int HbaseOp::scanclose()
{
	if (scanid < 0) {
		errnum = ERROR_CODE_NOT_OPENED; 
		DERROR("hbase scan close error: must open first\n");
		return -1;
	}
	try {
		hclient->scannerClose(scanid);
		scanid = -1;
	} catch (hbase::thrift::IOError &e) {
		errnum = ERROR_CODE_IOERROR;
        DERROR("hbase scan close error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
	} catch (hbase::thrift::IllegalArgument &e) {
		errnum = ERROR_CODE_ILLEGAL_ARGUMENT;
        DERROR("hbase scan close error: errno:%d:%d:%s\n", errno, errnum, e.what());
		return -1;
    } catch (TTransportException &e) {
        if (e.getType() == TTransportException::NOT_OPEN) {
            errnum = ERROR_CODE_SOCKET_CLOSED;
        } else {
            errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        }
        DERROR("hbase scan close error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;

    } catch (TException &e) {
        errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase scan close error: errno:%d:%d:%s\n", errno, errnum, e.what());
        return -1;
    }
/*
    } catch (...) {
		errnum = ERROR_CODE_DATA_STATUS_UNKNOWN;
        DERROR("hbase scan close error: %d\n", ERROR_CODE_DATA_STATUS_UNKNOWN);
        return -1;
    }
*/

	return 0;
}

int HbaseOp::open()
{
	if (hclient != NULL) {
		DERROR("hbase open error: already open\n");
		return 0;
	}
    socket.reset(new TSocket(sv_ip, sv_port));
    transport.reset(new TBufferedTransport(socket));
    protocol.reset(new TBinaryProtocol(transport));

	scanid = -1;
    hclient = new hbase::thrift::HbaseClient(protocol);
	try {
		transport->open();
	} catch (TException &e) {
		DERROR("hbase connect error: errrno:%d:%s\n", errno, e.what());
		return -1;
	}
    return 0;
}

int HbaseOp::close()
{
	if (hclient != NULL) {
		try {
			transport->close();
		} catch (TException &e) {
			DERROR("hbase close error: errrno:%d:%s\n", errno, e.what());
			return -1;
		}
		delete(hclient);
		hclient = NULL;
	}
	scanid = -1;
	return 0;
}

    void
free_rows(RowResult *rows)
{
    RowResult *per_row;
    RowResult *tmp;

    per_row = rows;

    while(per_row != NULL) {
        //rows = rows->next();
        tmp = per_row;
        per_row = tmp->next;
        tmp->free();
        delete(tmp);
    }
    return;
}

